package org.shuzhou.c_operator;

import java.time.LocalDate;
import java.time.Period;
import java.time.format.DateTimeFormatter;

import org.shuzhou.model.PersonalInfo;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 用到的数据集为data/people.txt
 */
public class PersonAnalyzer {
    public static void main(String[] args) throws Exception {

        // set up the streaming execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /**
         * 获取数据，数据源为Kafka，避免从端口获取中文数据乱码
         * 数据格式
         * 王矗馨,女,汉族,411424195001010028,19500101,150977754190,kk4jlw850x@sina.com,甘肃省临夏回族自治州和政县CNR路884号
         */
        KafkaSource<String> source = KafkaSource.<String>builder()
        .setBootstrapServers("192.168.56.151:9092,192.168.56.152:9092,192.168.56.153:9092")
        .setTopics("persion")
        .setGroupId("root")
        .setStartingOffsets(OffsetsInitializer.earliest())
        .setValueOnlyDeserializer(new SimpleStringSchema())
        .build();

        DataStream<String> people = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

        // map操作，提取数据中的民族
        // SingleOutputStreamOperator<String> res = people.map(new MapFunction<String,
        // String>(){
        // @Override
        // public String map(String value) throws Exception {
        // String nation = value.split(",")[2];
        // return nation;
        // }
        // });

        // res.print();

        /**
         * 获取数据，数据源为kafka端口
         * 数据格式
         * 王矗馨,女,汉族,411424195001010028,19500101,150977754190,kk4jlw850x@sina.com,甘肃省临夏回族自治州和政县CNR路884号
         */
        SingleOutputStreamOperator<PersonalInfo> ds = people.map(new MapFunction<String, PersonalInfo>() {
            @Override
            public PersonalInfo map(String value) throws Exception {
                // System.out.println(value.toString());
                String encode = new String(value.getBytes("UTF-8"), "UTF-8");
                String[] fields = encode.split(",");
                PersonalInfo info = new PersonalInfo(fields[0], fields[1], fields[2], fields[3], fields[4], fields[5],
                        fields[6], fields[7]);
                return info;
            }
        });

        // ds.print();

        /**
         * filter数据过滤，获取民族为汉族的数据
         */
        SingleOutputStreamOperator<PersonalInfo> filter = ds.filter(new FilterFunction<PersonalInfo>() {
            @Override
            public boolean filter(PersonalInfo persion) throws Exception {
                return persion.getEthnicity().equals("汉族") ;
            }
        });

        // lambda 表达式简化代码
        // SingleOutputStreamOperator<PersonalInfo> filter = ds.filter(persion -> persion.getGender().equals("汉族"));


        // filter.print();
        // 将结果输出到控制台
        // filter.addSink(new SinkFunction<PersonalInfo>() {
        // @Override
        // public void invoke(PersonalInfo personalInfo, Context context) throws Exception {
        //     String info = personalInfo.getName();
        //     System.out.println(new String(info.getBytes("UTF-8"), "UTF-8"));
        // }
        // });

        // filter.keyBy(persion -> persion.getEthnicity()).sum(1);
        

        /**
         * keyBy示例：按照民族进行分组
         */
        KeyedStream<PersonalInfo, String> keyBy = ds.keyBy(new KeySelector<PersonalInfo, String>(){
            @Override
            public String getKey(PersonalInfo product) throws Exception {
            return product.getEthnicity();
            }
        });

        /**
         * 先使用map计算每个人的年龄，得到元组(民族,age)
         * 然后使用keyBy对民族进行分组
         */
        KeyedStream<Tuple2<String, Integer>, Tuple> keyedStream = ds.map(new MapFunction<PersonalInfo, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(PersonalInfo personalInfo) throws Exception {
                long age = calculateAge(personalInfo.getDateOfBirth());
                return Tuple2.of(personalInfo.getEthnicity(), (int) age);
            }
        }).keyBy(0);
        

        /**
         * reduce聚合案例：对每个民族（已分组）的年龄进行累加，输出元组(民族，年龄总和)
         */
        keyedStream.reduce(new ReduceFunction<Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
                return Tuple2.of(value1.f0, value1.f1 + value2.f1);
            }
        }).print();


        /**
         * 年龄累加案例，可以使用Aggregations算子中的sum完成
         */
        // ds.map(new MapFunction<PersonalInfo, Tuple2<String, Integer>>() {
        //     @Override
        //     public Tuple2<String, Integer> map(PersonalInfo personalInfo) throws Exception {
        //         long age = calculateAge(personalInfo.getDateOfBirth());
        //         return Tuple2.of(personalInfo.getEthnicity(), (int) age);
        //     }
        // }).keyBy(0).sum(1).print();
        

        // KeyedStream<PersonalInfo, String> keyBy = ds.keyBy(value -> value.getEthnicity());


        env.execute("Flink Operator Example");
    }

    // 计算年龄的函数
    public static int calculateAge(String dateOfBirth) {
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd");
            LocalDate birthDate = LocalDate.parse(dateOfBirth, formatter);
            LocalDate currentDate = LocalDate.now();
            Period period = Period.between(birthDate, currentDate);
            return period.getYears();
        }
}
